package syM

import (
	"db2s/global"
	"db2s/ref"
	mq "db2s/topic-mq"
	"sync"
	"sync/atomic"
)

type indexMetaComparison struct {
	AddIndexName any `json:"add"`
	DelIndexName any `json:"del"`
}
type colMetaComparison struct {
	AddColName    any `json:"add"`
	DelColName    any `json:"del"`
	ModifyColName any `json:"modify"`
}
type partMetaComparison struct {
	AddColName any `json:"add"`
	DelColName any `json:"del"`
}
type getTableMeta struct {
	CommentMe       any                 `json:"commentMe"`
	Constraints     any                 `json:"constraintsMe"`
	CharacterMe     any                 `json:"characterMe"`
	CollationMe     any                 `json:"collationMe"`
	ColumnMe        any                 `json:"columnMe"`
	ColumnMetaMap   any                 `json:"columnMetaMap"`
	IndexMe         any                 `json:"indexMe"`
	PartMe          any                 `json:"partMe"`
	IndexComparison indexMetaComparison `json:"indexComparison"`
	ColComparison   colMetaComparison   `json:"colComparison"`
	PartComparison  partMetaComparison  `json:"partComparison"`
	AlterStatus     bool                //判断是否为create,true为alter，false为create
}

func structIsBaseQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var event = "[structIsBaseQueryListeners]"
	var plan = message.Properties["plan"].(*SchedulePlan2)
	var wg sync.WaitGroup
	defer func() {
		_ = ref.RecoverPanic(event, recover())
		baseMsgSendBody.MsgCurryDone()
		baseMsgSendBody.MsgCurryWgDone()
	}()
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": StructIsBaseQuery, "outFay": GlobalPConfigs.result.Teletypewriter})
	if metaGlobalInitActive(message) != nil {
		return mq.ReconsumeLater
	}
	plan.MQ.Wg = &wg
	atomic.SwapInt64(plan.MQ.FirstMsg.QuitExec, 1)
	plan.MQ.AbnormalQuit, plan.MQ.NormalQuit = global.EFunc{FuncName: mqStatusMonitorAbnormalQuit, Params: plan}, global.EFunc{FuncName: mqStatusMonitorNormalQuit, Params: plan}
	plan.MQ.MqTableStatusMonitor(getTableName(plan.Object.Schema, plan.Object.Table))
	if structBaseQuery(message) != nil {
		return mq.ReconsumeLater
	}
	wg.Wait()
	return mq.ConsumeSuccess
}

// structMetaListeners structMeta  监听执行器
func structIsColumnMetaQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var event = "structIsColumnMetaQueryListeners"
	defer func() {
		_ = ref.RecoverPanic(event, recover())
	}()
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		tableMeta1 = message.Properties["tableMeta1"].(*getTableMeta)
		tableMeta  = message.Properties["tableMeta"].(*global.GetTableMeta)
		err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": StructIsColumnMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	if structOptionsCheck("column") {
		if tableMeta1.ColumnMe, err = structColumnMetaData(message); err != nil {
			return mq.ReconsumeLater
		}
		if tableMeta1.ColumnMetaMap, err = structColumnMetaMapData(message, tableMeta1.ColumnMe); err != nil {
			return mq.ReconsumeLater
		}
	}
	if !sendMsg(getCheckMod(), StructIsIndexMeta, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "tableMeta": tableMeta, "tableMeta1": tableMeta1, "topic": StructIsIndexMeta,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func structIsIndexMetaQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var event = "structIsColumnMetaQueryListeners"
	defer func() {
		_ = ref.RecoverPanic(event, recover())
	}()
	var (
		err        error
		plan       = message.Properties["plan"].(*SchedulePlan2)
		tableMeta  = message.Properties["tableMeta"].(*global.GetTableMeta)
		tableMeta1 = message.Properties["tableMeta1"].(*getTableMeta)
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": StructIsIndexMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	if structOptionsCheck("index") {
		if tableMeta1.IndexMe, err = structIndexMetaData(message); err != nil {
			return mq.ReconsumeLater
		}
	}
	if !sendMsg(getCheckMod(), StructIsPartMeta, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "tableMeta": tableMeta, "tableMeta1": tableMeta1, "topic": StructIsPartMeta,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func structIsPartMetaQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var event = "structIsColumnMetaQueryListeners"
	defer func() {
		_ = ref.RecoverPanic(event, recover())
	}()
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		tableMeta1 = message.Properties["tableMeta1"].(*getTableMeta)
		err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": StructIsPartMeta, "outFay": GlobalPConfigs.result.Teletypewriter})
	if structOptionsCheck("partition") {
		if tableMeta1.PartMe, err = structPartitionMetaData(message); err != nil {
			return mq.ReconsumeLater
		}
	}
	if !sendMsg(getCheckMod(), StructIsMetaComparison, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "tableMeta": message.Properties["tableMeta"], "tableMeta1": message.Properties["tableMeta1"], "topic": StructIsMetaComparison,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func structIsMetaComparisonListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var event = "structIsColumnMetaQueryListeners"
	defer func() {
		_ = ref.RecoverPanic(event, recover())
	}()
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		tableMeta1 = message.Properties["tableMeta1"].(*getTableMeta)
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": StructIsMetaComparison, "outFay": GlobalPConfigs.result.Teletypewriter})
	if tableMeta1.AlterStatus { //是校验状态不是create table状态
		if metaComparison(tableMeta1, plan) != nil {
			return mq.ReconsumeLater
		}
	}
	if !sendMsg(getCheckMod(), StructIsFix, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "tableMeta1": message.Properties["tableMeta1"], "topic": StructIsFix,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func structIsFixListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var event = "structIsColumnMetaQueryListeners"
	defer func() {
		_ = ref.RecoverPanic(event, recover())
	}()
	var (
		plan       = message.Properties["plan"].(*SchedulePlan2)
		tableMeta1 = message.Properties["tableMeta1"].(*getTableMeta)
		result     []fixSqlResult
		err        error
	)
	defer deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": StructIsFix, "outFay": GlobalPConfigs.result.Teletypewriter})
	if tableMeta1.AlterStatus { //是校验状态不是create table状态
		if result, err = alterFixSql(plan, tableMeta1); err != nil {
			return mq.ReconsumeLater
		}
	} else {
		if result, err = createFixSql(plan, tableMeta1); err != nil {
			return mq.ReconsumeLater
		}
	}
	if !sendMsg(getCheckMod(), StructIsApply, getTableName(plan.Object.Schema, plan.Object.Table), getMsgSendData(plan, map[string]any{
		"logSeq": plan.Object.ObjectLogSeq, "fixMeta": message.Properties["tableMeta1"], "topic": StructIsApply, "fixSqlResult": result,
	})) {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
func structIsApplyListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
	var event = "structIsColumnMetaQueryListeners"
	defer func() {
		_ = ref.RecoverPanic(event, recover())
	}()
	var plan = message.Properties["plan"].(*SchedulePlan2)
	defer func() {
		deleteTopic(map[string]any{"subTask": plan.subTaskInfo, "topic": StructIsApply, "outFay": GlobalPConfigs.result.Teletypewriter})
		plan.MQ.FirstMsgCustomerAdd()
		plan.MQ.FirstObjectCustomerAdd()
		plan.MQ.SecondMsgCustomerAdd()
	}()
	if execFixSql(message) != nil {
		return mq.ReconsumeLater
	}
	return mq.ConsumeSuccess
}
